// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: leon

#include "coin2/exchange/mexc/feed_v2/processor.h"

#include <set>
#include <string_view>

#include <boost/algorithm/string.hpp>

#include "coin2/base/conversion.h"
#include "coin2/base/string_util.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/mexc/feed_v2/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::mexc::feed_v2::impl {
void MexcFeedProcessor::OnTopicRecord(const TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  const auto& data = packet.payload;
  rapidjson::Document document;
  document.Parse(data.c_str());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: "
            << GetParseError_En(document.GetParseError()) << packet.payload;
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  const auto& channel = ExtractChannel(document);
  if (!CheckChannelFilter(channel)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }
  if (document.HasMember("symbol")) {
    std::string native_symbol = document["symbol"].GetString();
    boost::erase_all(native_symbol, "_");

    if (channel == "book") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        MexcFeedUpdateParser::ParseBook(book, document["data"], packet.timestamp);
        upd_mgr_.PublishBook();
      }
    } else if (channel == "limited_depth") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        if (MexcFeedUpdateParser::ParseSnapshot(book, document["data"], packet.timestamp)) {
          upd_mgr_.PublishBook();
        }
      }
    } else if (channel == "incremental_depth") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        if (MexcFeedUpdateParser::ParseIncremental(book, document["data"], packet.timestamp)) {
          upd_mgr_.PublishBook();
        }
      }
    } else if (channel == "trade") {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }
      Trade* agg_trade = upd_mgr_.GetTradeMutable();
      agg_trade->Clear();
      Trade trade{};
      for (auto& trade_json : document["data"]["deals"].GetArray()) {
        MexcFeedUpdateParser::ParseTrade(trade_json, &trade);
        if (TryAggregateTrades(agg_trade, trade)) {
          continue;
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTrade(native_symbol, true);
        }
        *agg_trade = trade;
      }
      if (agg_trade->fill_qty > 0) {
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else {
      THROW() << "[Mexc Feed Processor] unkown topic: " << packet.payload;
    }
  } else if (channel == "kline_rest") {
    const std::string& url = document["url"].GetString();
    auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
    std::string native_symbol = parameters["symbol"];
    std::string kline_interval = parameters["interval"];
    auto interval = MexcFeedUpdateParser::ParseKlineInterval(kline_interval);
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines != nullptr) {
      const auto& data = document["data"].GetArray();
      int page = -1;
      if (parameters.count("page") > 0) {
        page = std::stoi(parameters["page"]);
      }
      if (page == 0) {
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      bool to_publish = false;
      for (const auto& kline_data : data) {
        bool freezed = MexcFeedUpdateParser::ParseKlineRest(kline_data,
            klines, interval, packet.timestamp);
        to_publish = to_publish || freezed;
      }
      if (to_publish) upd_mgr_.PublishKline(native_symbol);
    }
  } else {
    THROW() << "[Mexc Feed Processor] meta message: " << packet.payload;
  }
  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::mexc::feed_v2::impl
